/**
 * This file is part of Sonedyan.
 * 
 * Sonedyan is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation;
 * either version 3 of the License, or (at your option) any
 * later version.
 *
 * Sonedyan is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public
 * License along with Octave; see the file COPYING.  If not
 * see <http://www.gnu.org/licenses/>.
 * 
 * Copyright (C) 2009-2013 Jimmy Dubuisson <jimmy.dubuisson@gmail.com>
 */

package org.unige.mpej.eckmann.sonedyan.mr;

import it.unimi.dsi.io.ByteBufferInputStream;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

import org.apache.log4j.Logger;

import com.google.common.io.ByteSink;
// import com.google.common.io.ByteSource;
import com.google.common.io.ByteSource;
import com.google.common.io.FileWriteMode;
import com.google.common.io.Files;

public class LocalGraphLoader
{
	private static Logger log = Logger.getLogger(org.unige.mpej.eckmann.sonedyan.mr.LocalGraphLoader.class);

	private static final String GRAPH_FOLDER = "/home/jimmy/datasets/twitter/children/";
	protected static final String DATA_FILE_NAME = "graph.data";
	protected static final String INDEX_FILE_NAME = "graph.index";
	// private static final String VERTICES_FILE_NAME = "graph.vertices";
	
	// Twitter index and data size in int
	public static final int TWITTER_INDEX_SIZE = 41652230;
	public static final int TWITTER_DATA_SIZE = 1468365182;
	
	public HashMap<Integer, Integer> index = new HashMap<Integer, Integer>();
	public HashMap<Integer,Integer> nextVertex = new HashMap<Integer, Integer>();
	public int firstVertex;
	
	protected int numberOfEdges;
	
	public int[] data = new int[TWITTER_DATA_SIZE];
	//private ByteBufferInputStream data;
	
	private String dataFilePath;
	private String indexFilePath;

	/**
	 * constructor
	 */
	public LocalGraphLoader()
	{
		this.dataFilePath = GRAPH_FOLDER + DATA_FILE_NAME;
		this.indexFilePath = GRAPH_FOLDER + INDEX_FILE_NAME;
		this.loadGraph();
	}
	
	/**
	 * constructor
	 */
	public LocalGraphLoader(String indexFilePath, String dataFilePath)
	{
		this.indexFilePath = indexFilePath;
		this.dataFilePath = dataFilePath;
		this.loadGraph();
	}
	
	/**
	 * write edge list in text format to a custom binary format
	 * 
	 * data (4 bytes per vertex): children1, children2, ...
	 * index (4 bytes per vertex): vertex1 position1, ... 
	 * 
	 * NB: vertices of the edge list text file should be sorted
	 */
	public void writeGraph(String path)
	{
		try
		{
			BufferedReader br = new BufferedReader(new FileReader(path));
			
			File out1 = new File(this.dataFilePath);
			File out2 = new File(this.indexFilePath);
			// File out3 = new File(VERTICES_FILE_NAME);
			
			ByteSink sink1 = Files.asByteSink(out1, FileWriteMode.APPEND);
			BufferedOutputStream bws1 = sink1.openBufferedStream();
			
			ByteSink sink2 = Files.asByteSink(out2, FileWriteMode.APPEND);
			BufferedOutputStream bws2 = sink2.openBufferedStream();
			
			// ByteSink sink3 = Files.asByteSink(out3, FileWriteMode.APPEND);
			// BufferedOutputStream bws3 = sink3.openBufferedStream();
			
            String line;
            int currentVertex = -1;
            ArrayList<Integer> children = new ArrayList<Integer>();
            ArrayList<Integer> vertices = new ArrayList<Integer>();
            
            int pos = 0;
            int v1,v2;
            
            while ((line = br.readLine()) != null) 
            {
            	String[] es = line.split("\t");
            	
            	v1 = new Integer(es[0]);
            	v2 = new Integer(es[1]);

            	if (currentVertex == -1)
            	{
            		currentVertex = v1;
            		
            		bws2.write(intToByteArray(v1));
            		bws2.write(intToByteArray(0));
            		
            		vertices.add(v1);
            	}
            	
            	if (currentVertex != v1)
            	{
            		// complete index
            		bws2.write(intToByteArray(v1));
            		bws2.write(intToByteArray(pos));
            		
            		vertices.add(v1);
            		
            		// complete graph data
            		for (int i:children)
            		{
            			bws1.write(intToByteArray(i));
            		}

            		children = new ArrayList<Integer>();
            		currentVertex = v1;
            	}
            	
            	children.add(v2);
        		pos++;
            }
            
            // complete graph data
    		for (int i:children)
    		{
    			bws1.write(intToByteArray(i));
    		}
            
            log.info("# lines read: " + pos);
            
            /*
            for (int i: vertices)
            {
            	bws3.write(intToByteArray(i));
            }
            */
            
            br.close();
            bws1.close();
            bws2.close();
            // bws3.close();
		}
		catch(IOException e)
		{
			log.error("An error occured: " + e.getMessage());
		}
	}
	
	/**
	 * load binary graph in memory
	 */
	public void loadGraph()
	{
		try
		{
			log.info("Loading graph data");
			
			File in1 = new File(this.indexFilePath);
			
			// load index
			byte[] iba = Files.toByteArray(in1);
			int currentV = -1;
			
			for (int i = 0; i < iba.length / 8; i++)
			{
				int v = ((iba[0 + 8*i] & 0xFF) << 24) | ((iba[1 + 8*i] & 0xFF) << 16) | ((iba[2 + 8*i] & 0xFF) << 8) | (iba[3 + 8*i] & 0xFF);
				int pos = ((iba[4 + 8*i] & 0xFF) << 24) | ((iba[5 + 8*i] & 0xFF) << 16) | ((iba[6 + 8*i] & 0xFF) << 8) | (iba[7 + 8*i] & 0xFF);
				
				this.index.put(v, pos);
				
				if (currentV == -1)
				{
					currentV = v;
					this.firstVertex = v;
				}
				else
				{
					this.nextVertex.put(currentV, v);
					currentV = v;
				}
			}
			
			// add the last vertex
			this.nextVertex.put(currentV, null);
			
			log.info("Graph index successfully loaded");
			
			// empty array to free memory
			iba = new byte[0];
			
			// load data
			
			File in2 = new File(GRAPH_FOLDER + DATA_FILE_NAME);
			long dataLength = in2.length();
			
			ByteSource source = Files.asByteSource(in2);
			BufferedInputStream bis = source.openBufferedStream();
			
			byte[] bytes4 = new byte[4];
			
			int arraySize = (int) (dataLength/4);
			this.data = new int[arraySize];
			
			int counter = 0;
			
			while (bis.read(bytes4) != -1)
			{
				int v = ((bytes4[0] & 0xFF) << 24) | ((bytes4[1] & 0xFF) << 16) | ((bytes4[2] & 0xFF) << 8) | (bytes4[3] & 0xFF);
				this.data[counter] = v;
				counter++;
			}
			
			this.numberOfEdges = counter;
			
			bis.close();
			
			/*
			RandomAccessFile memoryMappedFile = new RandomAccessFile(this.dataFilePath, "r");
			this.data = ByteBufferInputStream.map(memoryMappedFile.getChannel(), FileChannel.MapMode.READ_ONLY);
			*/
			
			log.info("Graph data successfully loaded");
			
			log.info("# of vertices: " + this.nextVertex.size());
			log.info("# of edges: " + counter);
			log.info("# of indexes: " + this.index.size());
		}
		catch(IOException e)
		{
			log.error("An error occured: " + e.getMessage());
		}
	}
	
	/**
	 * get vertex children (byte array)
	 */
	/*
	public ArrayList<Integer> getChildren(int v)
	{
		try
		{
			int pos1 = this.index.get(v);
			Integer nextV = this.nextVertex.get(v);
			int size;
			
			ArrayList<Integer> children = new ArrayList<Integer>();
			byte[] bytes4 = new byte[4];
			
			if (nextV != null)
				size = this.index.get(nextV) - pos1;
			else
				size = (int) this.data.length()/4 - pos1;
			
			this.data.position((long) pos1*4);
			
			for (int i = 0; i < size; i++)
			{
				this.data.read(bytes4);
				children.add((int) byteArrayToLong(bytes4));
			}
			
			return children;
		}
		catch(IOException e)
		{
			log.error("Unable to read graph data: " + e.getMessage());
			return null;
		}
	}
	*/
	
	/**
	 * get vertex children
	 */
	public ArrayList<Integer> getChildren(int v)
	{
		try
		{
			ArrayList<Integer> children = new ArrayList<Integer>();
			
			// if vertex v is a sink, null will be returned
			Integer pos1;
			
			if ((pos1 = this.index.get(v)) != null)
			{
				Integer nextV = this.nextVertex.get(v);
				int size;
				
				if (nextV != null)
					size = this.index.get(nextV) - pos1;
				else
					size = this.numberOfEdges - pos1;
				
				for (int i = 0; i < size; i++)
				{
					// vertices are set to -1 when they get removed from the graph
					int c = this.data[pos1 + i];
					
					if (c != -1)
						children.add(c);
				}
			}
	
			return children;
		}
		catch(Exception e)
		{
			log.error("Unable to get children: " + e.getMessage());
			return null;
		}
	}
	
	/**
	 * remove vertex children
	 */
	public boolean removeSourceVertexChildren(int v)
	{
		try
		{
			// if vertex v is a sink, null will be returned
			Integer pos1;
			
			if ((pos1 = this.index.get(v)) != null)
			{
				Integer nextV = this.nextVertex.get(v);
				int size;
				
				if (nextV != null)
					size = this.index.get(nextV) - pos1;
				else
					size = this.numberOfEdges - pos1;
				
				for (int i = 0; i < size; i++)
				{
					// vertices are set to -1 when they get removed from the graph
					this.data[pos1 + i] = -1;
				}
			}
			
			return true;
		}
		catch(Exception e)
		{
			log.error("Unable to remove children: " + e.getMessage());
			return false;
		}
	}
	
	/**
	 * remove index entries if they have no child left
	 */
	public Set<Integer> getDanglingIndices()
	{
		Set<Integer> dangling = new HashSet<Integer>();
		
		for (int i: this.index.keySet())
		{
			ArrayList<Integer> children = this.getChildren(i);
			if (children != null && children.size() == 0)
			{
				dangling.add(i);
			}
		}
		
		return dangling;
	}
	
	/**
	 * main method
	 */
	public static void main(String args[])
	{
		/*
		// write graph binary adjacency list
		writeGraph(args[0]);
		*/
		
		// load graph binary data
		LocalGraphLoader loader = new LocalGraphLoader();
		
		// get children of node 12
		ArrayList<Integer> children = loader.getChildren(12);
	
		System.out.println("# indexes: " + loader.index.size());
		System.out.println("Index of vertex 140: " + loader.index.get(140));
		
		System.out.println("# children of vertex 12: " + children.size());
		
		/*
		Iterator<Integer> it = children.iterator();
		
		while (it.hasNext())
			System.out.println("Vertex child: " + it.next());
		*/
	}
	
	/**
	 * get string representation of specified byte buffer
	 */
	public static String byteArrayToString(byte[] bytes)
	{
		String s = "";
		
		for (byte b : bytes)
		{
			   s += String.format("0x%x ", b);
		}
		
		return s;
	}
	
	/**
	 * convert an int to a byte array
	 */
	public static byte[] intToByteArray(int value)
	{
	    return new byte[] {
	            (byte)(value >>> 24),
	            (byte)(value >>> 16),
	            (byte)(value >>> 8),
	            (byte)value};
	}
	
	/**
	 * Converts a 4 byte array of unsigned bytes to an long
	 * 
	 * @param b an array of 4 unsigned bytes
	 * @return a long representing the unsigned int
	 */
	public static long byteArrayToLong(byte[] b) 
	{
	    long l = 0;
	    l |= b[0] & 0xFF;
	    l <<= 8;
	    l |= b[1] & 0xFF;
	    l <<= 8;
	    l |= b[2] & 0xFF;
	    l <<= 8;
	    l |= b[3] & 0xFF;
	    return l;
	}
	
	/**
	 * char array to byte array
	 */
	public static byte[] charArrayToByteArray(char[] ca)
	{
		byte[] b = new byte[ca.length << 1];
		for(int i = 0; i < ca.length; i++) 
		{
			int bpos = i << 1;
			b[bpos] = (byte) ((ca[i]&0xFF00)>>8);
			b[bpos + 1] = (byte) (ca[i]&0x00FF);
		}
		return b;
	}
	
}
